package com.ds.infrastructure.hyperspace.container.endpoint.destination;

import com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.http.NettyHttpHeaderOptimizer;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.ReferenceCounted;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import static com.ds.infrastructure.hyperspace.common.StateEnum.*;

/**
 * @author  Weishaoying
 */
@Slf4j
public class Http1TargetProtocolAdapter extends AbstractTargetProtocolAdapter {

    private final NettyTargetEndpointChannelInitializer channelInitializer;

    private HyperspaceDuplexFlowContext currentDuplexFlowContext;

    private boolean connectionClose;

    public Http1TargetProtocolAdapter(NettyTargetEndpointChannelInitializer channelInitializer,
            NettyContainer nettyContainer,HyperspaceConnector hyperspaceConnector,
            TargetEndpointNode targetEndpointNode) {
        super(nettyContainer, hyperspaceConnector, targetEndpointNode);
        this.channelInitializer = channelInitializer;
    }

    @Override
    public void handleChannelRead(ChannelHandlerContext ctx, Object msg) {
        stream.touch();
        if (currentDuplexFlowContext == null || currentDuplexFlowContext.isClosed()) {
            ReferenceCountUtil.release(msg);
            return;
        }
        currentDuplexFlowContext.touch();
        if (msg instanceof HttpResponse) {
            HttpResponse httpResponse = (HttpResponse)msg;
            if (!HttpUtil.isKeepAlive(httpResponse)) {
                connectionClose = true;
            }
            currentDuplexFlowContext.setTargetResponseState(msg, httpResponse.status().code());
        }

        targetFlowOperation.handleOutput(currentDuplexFlowContext, msg);

        if (msg instanceof LastHttpContent) {
            currentDuplexFlowContext.touch();
            currentDuplexFlowContext.setOutputComplete(true);
            targetFlowOperation.handleOutputComplete(currentDuplexFlowContext);
            if (!currentDuplexFlowContext.getEndpointProtocol().isLongLive()) {
                currentDuplexFlowContext = null;
                if (!connectionClose) {
                    stream.releaseStreamWriteOperation();
                } else if(ctx.channel().isOpen()) {
                    ctx.close();
                }
            }
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        if (currentDuplexFlowContext != null) {
            currentDuplexFlowContext.touch();
            targetFlowOperation.handleOutputComplete(currentDuplexFlowContext);
        }
    }

    @Override
    public void switchProtocol(HyperspaceDuplexFlowContext duplexFlowContext,
            EndpointProtocolEnum protocol) {
        channelInitializer.switchProtocolFromHttp(stream, duplexFlowContext,
                channel.pipeline(), protocol);
    }

    @Override
    public void writeDataToTarget(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        if (duplexFlowContext.isPending()) {
            return;
        }
        if (data instanceof HttpRequest) {
            if (!channel.isActive()) {
                if (duplexFlowContext.getEndpointProtocol().isLongLive()) {
                    String message=String.format("Write data to channel failed! target node(%s),remote address:%s",
                    targetEndpointNode.getKey(),channel.remoteAddress());
                    targetFlowOperation.interruptAndCloseContext(duplexFlowContext,
                            TARGET_ENDPOINT_UNAVAILABLE, message);
                } else {
                    currentDuplexFlowContext = null;
                    hyperspaceConnector.onContextWriteFailed(duplexFlowContext);
                }
                return;
            }
            NettyHttpHeaderOptimizer.optimizeHttpHeader(duplexFlowContext, (HttpRequest) data, targetEndpointNode);
            duplexFlowContext.setTargetRemoteAddress(targetRemoteAddress);
            if(duplexFlowContext.getTracerContext()!=null){
                duplexFlowContext.getTracerContext().inject(targetRemoteAddress,(HttpRequest) data);
            }
        }
        channelHandlerContext.write(data).addListener(new WriteFutureListener(duplexFlowContext));
    }

    @Override
    public void clearSuspendData() {

    }

    @Override
    public HyperspaceDuplexFlowContext connectionDuplexFlowContext() {
        return currentDuplexFlowContext;
    }

    @Override
    public boolean requireDuplexFlowContext(HyperspaceDuplexFlowContext duplexFlowContext) {
        if (currentDuplexFlowContext == null) {
            currentDuplexFlowContext = duplexFlowContext;
            return true;
        }
        if (currentDuplexFlowContext != duplexFlowContext) {
            targetFlowOperation.interruptAndCloseContext(duplexFlowContext, GATEWAY_ERROR,
                    "DuplexFlowContext state incorrect, target node " +
                            targetEndpointNode.getKey());
            return false;
        }
        return true;
    }

    @Override
    public void checkTimeoutDuplexFlowContext() {
        if (currentDuplexFlowContext == null || currentDuplexFlowContext.isClosed()) {
            return;
        }
        if (currentDuplexFlowContext.checkTimeout()) {
            currentDuplexFlowContext.setTargetResponseState(
                    null, REQUEST_TIMEOUT.getHttpCode());
            currentDuplexFlowContext.onTimeout();
        }
    }

    private class WriteFutureListener implements GenericFutureListener<ChannelFuture> {

        private final HyperspaceDuplexFlowContext writeDuplexFlowContext;

        private WriteFutureListener(HyperspaceDuplexFlowContext duplexFlowContext) {
            this.writeDuplexFlowContext = duplexFlowContext;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                log.error("Write data to target channel failed!", future.cause());
                targetFlowOperation.interruptAndCloseContext(writeDuplexFlowContext,
                        TARGET_ENDPOINT_UNAVAILABLE,
                        String.format("Write data to channel failed!target node(%s),remote address:%s",
                                targetEndpointNode.getKey(), future.channel().remoteAddress()));
            }
        }
    }

    @Override
    public void releaseData(Object data) {
        if (data instanceof ReferenceCounted) {
            ReferenceCounted referenceCounted = (ReferenceCounted) data;
            if (referenceCounted.refCnt() > 0) {
                referenceCounted.release();
            }
        }
    }
}
